﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;

using STSdb.MTL.Utils.Serialization;

using STSdb.Data;

namespace STSdb.MTL
{
    public class TransactionsCoordinator
    {
        StorageEngine _storageEngine = null;
        SchemaCoordinator _schemaCoordinator = null;
        TablesCoordinator _tablesCoordinator = null;

        ReadLocatorsHandler _readLocatorsHandler = null;

        EngineTransactionsLog _engineTransactionLog = null;

        /// <summary>
        /// MAIN _SYNC FOR ALL COMMIT; ROLLBACK, SCHEMA OPEN, DELETE CREATE OPERATIONS
        /// </summary>
        public object _SYNC_COMMIT = new object();
               

        public TransactionsCoordinator(StorageEngine storageEngine)
        {
            this._storageEngine = storageEngine;

            this._schemaCoordinator = new SchemaCoordinator(this);

            this._tablesCoordinator = new TablesCoordinator(this);

            this._readLocatorsHandler = new ReadLocatorsHandler(this);
        }

        public StorageEngine StorageEngine
        {
            get { return this._storageEngine; }
        }

        public EngineTransactionsLog TransactionsLog
        {
            get { return this._engineTransactionLog; }
            set { this._engineTransactionLog = value; }
        }

        public SchemaCoordinator SchemaCoordinator
        {
            get { return this._schemaCoordinator; }
        }

        public TablesCoordinator TablesCoordinator
        {
            get { return this._tablesCoordinator; }
        }

        public ReadLocatorsHandler ReadLocatorsHandler
        {
            get { return this._readLocatorsHandler; }
        }


        #region "DB is operatable"
        /// <summary>
        /// No commit or rollback operations are possible if it's false
        /// </summary>
        private bool _dbIsOperatable = true;
        ReaderWriterLockSlim _sync_dbIsOperatable = new ReaderWriterLockSlim();

        public bool DbIsOperatable
        { 
            get 
            {
                _sync_dbIsOperatable.EnterReadLock();
                try
                {
                    return _dbIsOperatable;
                }
                finally
                {
                    _sync_dbIsOperatable.ExitReadLock();
                }
            }
            set 
            {

                _sync_dbIsOperatable.EnterWriteLock();
                try
                {
                    _dbIsOperatable = value;
                }
                finally
                {
                    _sync_dbIsOperatable.ExitWriteLock();
                }

            }
        }
        #endregion




        #region "Creating ans supporting Transactions"

        /// <summary>
        /// Key is transaction threadId
        /// </summary>
        Dictionary<int, TransactionUnit> _transactions = new Dictionary<int, TransactionUnit>();
        ReaderWriterLockSlim _sync_transactions = new ReaderWriterLockSlim();

        public Transaction GetTransaction()
        {
            if (!this.DbIsOperatable)
                return null;

            //Transaction must have 2 classes one class is for the user, with appropriate methods, second for technical purposes TransactionDetails, where we store different transaction information
            //both classes must be bound into one class TransactionUnit

            TransactionUnit transactionUnit = new TransactionUnit(this);

            //Checking if the same transaction already exists in the list of Transactions. 
            //It could happen in case of abnormal termination of parallel thread, without disposing of the transaction.
            //So we delete pending transaction first, then create new one.
            bool reRun = false;
            _sync_transactions.EnterReadLock();
            try
            {
                if (this._transactions.ContainsKey(transactionUnit.TransactionThreadId))
                {
                    reRun = true;
                }
            }
            finally
            {
                _sync_transactions.ExitReadLock();
            }

            if (reRun)
            {
                UnregisterTransaction(transactionUnit.TransactionThreadId);
                return GetTransaction();
            }


            _sync_transactions.EnterWriteLock();
            try
            {             

                this._transactions.Add(transactionUnit.TransactionThreadId, transactionUnit);
            }
            catch (System.Exception ex)
            {
                throw ex;
            }
            finally
            {
                _sync_transactions.ExitWriteLock();
            }

            return transactionUnit.Transaction;           
        }

        private TransactionUnit GetTransactionUnit(int transactionThreadId)
        {
            TransactionUnit transactionUnit = null;

            _sync_transactions.EnterReadLock();
            try
            {
                this._transactions.TryGetValue(transactionThreadId, out transactionUnit);
            }
            catch (System.Exception ex)
            {
                throw ex;
            }
            finally
            {
                _sync_transactions.ExitReadLock();
            }

            return transactionUnit;
        }

        public void UnregisterTransaction(int transactionThreadId)
        {
            TransactionUnit transactionUnit = this.GetTransactionUnit(transactionThreadId);

            if (transactionUnit != null)
            {
                _sync_transactions.EnterWriteLock();
                try
                {
                    transactionUnit.Dispose();
                    this._transactions.Remove(transactionUnit.Transaction.ManagedThreadId);

                    //Cleaning ReadLocatorsHandler, who registers all read locators
                    this.ReadLocatorsHandler.TransactionIsFinished(transactionThreadId);
                }
                catch (System.Exception ex)
                {
                    throw ex;
                }
                finally
                {
                    _sync_transactions.ExitWriteLock();
                }                
            }

            //letting other threads, which tried to register tables for modification and were blocked, to re-try the operation.
            mreWriteTransactionLock.Set();
        }

        /// <summary>
        /// Is called by STSdbMTLEngine on Dispose.
        /// </summary>
        public void UnregisterAllTransactions()
        {
            _sync_transactions.EnterWriteLock();
            try
            {
                foreach (var transactionUnit in _transactions.Values)
                {
                    transactionUnit.Dispose();                    
                }

                this._transactions.Clear();
            }
            catch (System.Exception ex)
            {
                throw ex;
            }
            finally
            {
                _sync_transactions.ExitWriteLock();
            }

            //lettign other threads, which tried to register tables for modification and were blocked, to re-try the operation.
            mreWriteTransactionLock.Set();
        }



        #region "Registering Tables for Writing or Read-Commited before making operations, for avoiding deadLocks"

        System.Threading.ManualResetEvent mreWriteTransactionLock = new System.Threading.ManualResetEvent(true);
        object _sync_dl = new object();

        /// <summary>
        /// Access synchronizer.
        /// All calls of the WRITE LOCATOR come over this function.
        /// </summary>
        /// <param name="transactionThreadId"></param>
        /// <param name="tablesNames"></param>
        public void RegisterWriteTablesForTransaction(int transactionThreadId, List<string> tablesNames)
        {
            //in every transaction unit we got a list of reserved for WRITING tables
            
            //if we have in tablesNames one of the tables which is in this list we have to stop the thread with mre
            bool toWaitTillTransactionIsFinished = false;
            bool breakOuterLoop = false;
            TransactionUnit transactionUnit = null;
            bool deadlock = false;

            while (true)    //loop till thread will get full access to write tables
            {
                toWaitTillTransactionIsFinished = false;
                breakOuterLoop = false;
                deadlock = false;

                //only tables required for writing or read-commited will have to go over this fast bottleneck
                lock (_sync_dl) 
                {
                    _sync_transactions.EnterReadLock();
                    try
                    {

                        this._transactions.TryGetValue(transactionThreadId, out transactionUnit);

                        if (transactionUnit == null)
                            return; //transaction doesn't exist anymore

                        //we check here if our tables (inside transaction) already marked as reserved, then we don't need other checks.                        
                        if (tablesNames.Except(transactionUnit.GetTransactionWriteTablesNames()).Count() == 0)
                            return;


                        //iterating over all open transactions except self, finding out if desired tables are locked by other threads.
                        foreach (var tu in this._transactions.Where(r => r.Value.TransactionThreadId != transactionThreadId))
                        {
                            foreach (string tableName in tu.Value.GetTransactionWriteTablesNames())
                            {
                                if (tablesNames.Contains(tableName))
                                {
                                    //
                                    //++++++++++++++ here we can register all tables which are waiting for write lock release
                                    transactionUnit.AddTransactionWriteTablesAwaitingReservation(tablesNames);

                                    //++++++++++++++ if thread, who has locked this table has another table in a "waiting for reservation" blocked by this thread - it's a deadlock                                    
                                    if (transactionUnit.GetTransactionWriteTablesNames().Intersect(tu.Value.GetTransactionWriteTablesAwaitingReservation()).Count() > 0)
                                    {
                                        //we got deadlock, we will stop this transaction with an exception
                                        deadlock = true;
                                    }

                                    //other thread has reserved table for the transaction we have to wait
                                    toWaitTillTransactionIsFinished = true;
                                    breakOuterLoop = true;

                                    if (!deadlock)
                                    {
                                        mreWriteTransactionLock.Reset();   //setting to signalled only in non-deadlock case
                                    }

                                    break;
                                }
                            }

                            if (breakOuterLoop)
                                break;
                        }
                    }
                    catch (System.Exception ex)
                    {
                        this.UnregisterTransaction(transactionThreadId);
                        throw new Exception("STSdb.MTL: Transaction COORD: "+ex.ToString());  
                    }
                    finally
                    {
                        _sync_transactions.ExitReadLock();
                    }                    

                    //if(true) this thread owns all table for modification lock
                    if (!toWaitTillTransactionIsFinished)
                    {
                        //+++++++++++ Here we can clear all table names in the waiting reservation queue
                        transactionUnit.ClearTransactionWriteTablesAwaitingReservation(tablesNames);

                        //we have to reserve for our transaction all tables
                        foreach (var tbn in tablesNames)
                            transactionUnit.AddTransactionWriteTable(tbn, null);

                        return;
                    }

                }//end of lock


                if (deadlock)
                {
                    this.UnregisterTransaction(transactionThreadId);
                    throw new Exception("STSdb.MTL: Transaction is in a deadlock, will be terminated, use RegisterWriteOrReadCommitedTables method to avoid deadlocks!");                    
                }

                if (toWaitTillTransactionIsFinished)
                {
                    //blocking thread which requires busy tables for writing, till they are released
                    mreWriteTransactionLock.WaitOne();
                }
            }
        }


        /// <summary>
        /// This check concerns only a user try to register tables for mustation.
        /// This Registration can be done only once for now and user shouldn't have any registered tables for mutations before.
        /// Probably logic can be revisited to let user make registrations more then once, 
        /// it can be possible if Rollback cleans all registrations, but it's dangerous is user doesn't know it
        /// and due to incorrect usage can cause deadlock after Rollback. May be to make two types of Rollback(bool clearReservedTables) or so...
        /// </summary>
        /// <param name="transactionThreadId"></param>
        /// <param name="tablesNames"></param>
        /// <returns></returns>
        public bool IfUserStillCanRegisterMutatableTables(int transactionThreadId, List<string> tablesNames)
        {
            TransactionUnit transactionUnit = null;

            _sync_transactions.EnterReadLock();
            try
            {
                        
                this._transactions.TryGetValue(transactionThreadId, out transactionUnit);

                if (transactionUnit == null)
                    return false;

                if (transactionUnit.GetTransactionWriteTablesNames().Count() > 0)
                    return false;

                return true;
            }
            catch (System.Exception ex)
            {
                throw ex;
            }
            finally
            {
                _sync_transactions.ExitReadLock();
            } 
        }
        #endregion //Eliminating Deadlocks. Registering tables for write before starting transaction operations


        #endregion //Creating ans supporting Transactions





        /// <summary>
        /// Returns table entity which can be mutated
        /// </summary>
        /// <typeparam name="T"></typeparam>
        /// <typeparam name="T1"></typeparam>
        /// <param name="tableName"></param>
        /// <param name="transactionThreadId"></param>
        /// <returns></returns>
        public XTableMTL<T, T1> Get_Table_Write<T, T1>(string tableName, int transactionThreadId)
        {
            TransactionUnit transactionUnit = this.GetTransactionUnit(transactionThreadId);

            if (transactionUnit != null)
            {
                if (System.Threading.Thread.CurrentThread.ManagedThreadId != transactionThreadId)
                {
                    this.UnregisterTransaction(transactionThreadId);
                    throw new Exception("STSdb.MTL: ONE TRANSACTION CAN BE USED ONLY FROM ONE THREAD");
                }


                //We must put Get_Table_Write through the same bottleneck as RegisterWriteTablesForTransaction
                this.RegisterWriteTablesForTransaction(transactionThreadId, new List<string> { tableName });
                //it will wait here till table for writing, reserved by other thread is released

                TableLocators<T, T1> tbl = _tablesCoordinator.GetTable<T, T1>(tableName);

                if (tbl == null)
                {
                    this.UnregisterTransaction(transactionThreadId);
                    throw new Exception("STSdb.MTL: CAN'T GET TABLE " + tableName);
                }
                else
                {
                    transactionUnit.AddTransactionWriteTable(tableName, tbl);    //added together with ITransactable
                    return new XTableMTL<T,T1>(tbl.GET_WRITE_TABLE(),tableName,this);
                }
            }
            else
            {
                throw new Exception("STSdb.MTL: TRANSACTION DOESN'T EXIST ANY MORE");
            }
        }


        /// <summary>
        /// Returns table entity which can't be mutated
        /// </summary>
        /// <typeparam name="T"></typeparam>
        /// <typeparam name="T1"></typeparam>
        /// <param name="tableName"></param>
        /// <param name="transactionThreadId"></param>
        /// <returns></returns>
        public XTableMTL<T, T1> Get_Table_Read<T, T1>(string tableName, int transactionThreadId)
        {

            TransactionUnit transactionUnit = this.GetTransactionUnit(transactionThreadId);

            if (transactionUnit != null)
            {
                if (System.Threading.Thread.CurrentThread.ManagedThreadId != transactionThreadId)
                {
                    this.UnregisterTransaction(transactionThreadId);
                    throw new Exception("STSdb.MTL: ONE TRANSACTION CAN BE USED ONLY FROM ONE THREAD");
                }

                //_LGC_ Registering starting of READ Locator work. Read ReadLocatorHasStartedItsJob comment for further info.
                //Read locator will be unregistered in the end of the transaction
                this.ReadLocatorsHandler.ReadLocatorStarted(tableName, transactionThreadId);

                TableLocators<T, T1> tbl = _tablesCoordinator.GetTable<T, T1>(tableName);

                if (tbl == null)
                {
                    this.UnregisterTransaction(transactionThreadId);
                    throw new Exception("STSdb.MTL: CAN'T GET TABLE " + tableName);
                }
                else
                {
                    return new XTableMTL<T, T1>(tbl.GetReadTableLocator(transactionThreadId), tableName, this);
                }
            }
            else
            {
                throw new Exception("STSdb.MTL: TRANSACTION DOESN'T EXIST ANY MORE");
            }

        }


        ///// <summary>
        ///// Returns table entity which can be mutated
        ///// </summary>
        ///// <typeparam name="T"></typeparam>
        ///// <typeparam name="T1"></typeparam>
        ///// <param name="tableName"></param>
        ///// <param name="transactionThreadId"></param>
        ///// <returns></returns>
        //public XTable<T,T1> Get_Table_Write<T,T1>(string tableName, int transactionThreadId)
        //{
        //    TransactionUnit transactionUnit = this.GetTransactionUnit(transactionThreadId);
          
        //    if (transactionUnit != null)
        //    {
        //        if (System.Threading.Thread.CurrentThread.ManagedThreadId != transactionThreadId)
        //        {
        //            this.UnregisterTransaction(transactionThreadId);
        //            throw new Exception("STSdb.MTL: ONE TRANSACTION CAN BE USED ONLY FROM ONE THREAD");
        //        }


        //        //We must put Get_Table_Write through the same bottleneck as RegisterWriteTablesForTransaction
        //        this.RegisterWriteTablesForTransaction(transactionThreadId, new List<string> { tableName });
        //        //it will wait here till table for writing, reserved by other thread is released

        //        TableLocators<T,T1> tbl = _tablesCoordinator.GetTable<T, T1>(tableName);

        //        if (tbl == null)
        //        {
        //            this.UnregisterTransaction(transactionThreadId);
        //            throw new Exception("STSdb.MTL: CAN'T GET TABLE " + tableName);                   
        //        }
        //        else
        //        {
        //            transactionUnit.AddTransactionWriteTable(tableName,tbl);    //added together with ITransactable
        //            return tbl.GET_WRITE_TABLE();
        //        }
        //    }
        //    else
        //    {
        //        throw new Exception("STSdb.MTL: TRANSACTION DOESN'T EXIST ANY MORE");
        //    }            
        //}

        ///// <summary>
        ///// Returns table entity which can't be mutated
        ///// </summary>
        ///// <typeparam name="T"></typeparam>
        ///// <typeparam name="T1"></typeparam>
        ///// <param name="tableName"></param>
        ///// <param name="transactionThreadId"></param>
        ///// <returns></returns>
        //public XTable<T, T1> Get_Table_Read<T, T1>(string tableName, int transactionThreadId)
        //{

        //    TransactionUnit transactionUnit = this.GetTransactionUnit(transactionThreadId);

        //    if (transactionUnit != null)
        //    {
        //        if (System.Threading.Thread.CurrentThread.ManagedThreadId != transactionThreadId)
        //        {
        //            this.UnregisterTransaction(transactionThreadId);
        //            throw new Exception("STSdb.MTL: ONE TRANSACTION CAN BE USED ONLY FROM ONE THREAD");
        //        }

        //        //_LGC_ Registering starting of READ Locator work. Read ReadLocatorHasStartedItsJob comment for further info.
        //        //Read locator will be unregistered in the end of the transaction
        //        this.ReadLocatorsHandler.ReadLocatorStarted(tableName, transactionThreadId);

        //        TableLocators<T, T1> tbl = _tablesCoordinator.GetTable<T, T1>(tableName);

        //        if (tbl == null)
        //        {
        //            this.UnregisterTransaction(transactionThreadId);
        //            throw new Exception("STSdb.MTL: CAN'T GET TABLE " + tableName);
        //        }
        //        else
        //        {                    
        //            return tbl.GET_READ_TABLE();
        //        }
        //    }
        //    else
        //    {
        //        throw new Exception("STSdb.MTL: TRANSACTION DOESN'T EXIST ANY MORE");
        //    }                      

        //}

        

        #region "Distributed Transactions. Transactions between multiple STSdb Engine instances."

        XTable<long, string> transactionInfoTable = null;
        object lock_transactionInfoTable = new object();

        /// <summary>
        /// Distributed Transaction. 
        /// </summary>
        /// <param name="transactionUnit"></param>
        public void DistributedCommitStart(TransactionUnit transactionUnit, long transactionId)
        {
            
            try
            {
                DstTranInfo dtInfo = new DstTranInfo()
                {
                    TransactionId = transactionId
                };

                foreach (var wt in transactionUnit.GetTransactionWriteTables())
                {
                    wt.SnapshotReadLocator(transactionId);

                    dtInfo.TransactionTables.Add(wt.TableName);
                }


                //Saving transaction information

                lock (lock_transactionInfoTable) //LOCKS MULTI THREADING ACCESS TO DISTRIBUTED TRANSACTIONS INFORMATION
                {
                    if (transactionInfoTable == null)
                        transactionInfoTable = this._schemaCoordinator.OpenTable<long, string>("@sys_dstTranInfo");

                    transactionInfoTable[transactionId] = dtInfo.SerializeXml();
                    transactionInfoTable.Commit();
                }


                ////////////////////////////////////////   NATIVE STSdb COMMIT, without changing of READ LOCATOR
                _sync_dbIsOperatable.EnterReadLock();
                try
                {
                    if (!this._dbIsOperatable)
                        return;

                    STSdb.Data.Transaction internalTransaction = new STSdb.Data.Transaction(this._storageEngine, transactionUnit.GetTransactionWriteTables().Select(r => r.STSdbITransaction).ToArray());

                    //Commit and Rollback must be serial for all threads, though commits are fast and take the same time, independent from the quantity of committed data.
                    lock (_SYNC_COMMIT)
                    {
                        internalTransaction.Commit();
                    }
                }
                catch (Exception ex)
                {
                    this.UnregisterTransaction(transactionUnit.TransactionThreadId);
                    throw new Exception("STSdb.MTL.TransactionsCoordinator.SnapshotInTransactionWriteTables: NATIVE COMMIT FAILED " + ex.ToString());
                }
                finally
                {
                    _sync_dbIsOperatable.ExitReadLock();
                }

            }
            catch (Exception ex)
            {
                throw new Exception("STSdb.MTL.TransactionsCoordinator.SnapshotInTransactionWriteTables: " + ex.ToString());
            }
        }

        /// <summary>
        /// Recreates Read locators from Write Locators, only in-memory operations.
        /// </summary>
        /// <param name="transactionUnit"></param>
        public void DistributedCommitEnd(TransactionUnit transactionUnit)
        {
            //internal transaction commit
            try
            {
                transactionUnit.Commit();
            }
            catch (Exception ex)
            {
                this.StopEngineInFailureCase("In Memory Synchronization failed in Commit. \r\n" + ex.ToString());
                return;
            }
        }

        /// <summary>
        /// Is called in case if one of the engines transactions failed to be executed
        /// </summary>
        /// <param name="transactionUnit"></param>
        /// <param name="transactionId"></param>
        public void DistributedRollback(TransactionUnit transactionUnit, long transactionId)
        {
            //Restoring Read and Write locators for every table inside of transaction from Snapshot
            foreach (var wt in transactionUnit.GetTransactionWriteTables())
            {
                wt.RestoreLocatorsFromSnapshot(transactionId);
            }

            //If all is done OK, then we delete inside of this engine information sys_dstTranInfo, it means, we don't need to do anything with this tables after engine restarts after failure.
            lock (lock_transactionInfoTable)
            {
                if (transactionInfoTable == null)
                    transactionInfoTable = this._schemaCoordinator.OpenTable<long, string>("@sys_dstTranInfo");

                transactionInfoTable.Delete(transactionId);
                transactionInfoTable.Commit();
            }
        }

               


        /// <summary>
        /// This function can be called only from STSdbMTLEngine constructor.
        /// Its primary goal is to recover tables, participating in failed distributed transacations, from their snapshots.
        /// List of failedDistributedTransactionsIds will be ordered in descending sequence
        /// </summary>
        /// <param name="failedDistributedTransactionsIds"></param>
        public void RestoreTablesFromSnapshots(List<long> failedDistributedTransactionsIds)
        {
            if (transactionInfoTable == null)
                transactionInfoTable = this._schemaCoordinator.OpenTable<long, string>("@sys_dstTranInfo");

            DstTranInfo transactionInfo=null;

            foreach (var transactionId in failedDistributedTransactionsIds.OrderByDescending(r=>r))
            {
                transactionInfo = transactionInfoTable[transactionId].DeserializeXml<DstTranInfo>();

                if (transactionInfo != null)
                {
                    foreach (var transactionTable in transactionInfo.TransactionTables)
                    {
                        this._schemaCoordinator.RestoreTableFromSnapshot(transactionTable, "@sys_trs_" + transactionId.ToString() + "_" + transactionTable);
                    }
                }

                //deleting failed transaction info table
                lock (lock_transactionInfoTable)
                {                    
                    transactionInfoTable.Delete(transactionId);
                    transactionInfoTable.Commit();
                }
            }
        }

        
        #endregion


        


        /// <summary>
        /// Commits transaction.
        /// If transaction ignore then we don't put saving tables via transaction mechanizm
        /// </summary>
        /// <param name="transactionThreadId"></param>
        /// <param name="ignoreTransaction">default is false</param>
        public void Commit(int transactionThreadId,bool ignoreTransaction)
        {
          
            TransactionUnit transactionUnit = this.GetTransactionUnit(transactionThreadId);

            if (transactionUnit != null)
            {

                ////////////////////////////////////////   NATIVE STSdb COMMIT
                _sync_dbIsOperatable.EnterReadLock();
                try
                {
                    if (!this._dbIsOperatable)
                        return;

                    ITransaction[] tablesForTransaction = transactionUnit.GetTransactionWriteTables().Select(r => r.STSdbITransaction).ToArray();
                    
                    if (tablesForTransaction.Count() == 0)
                    {

                    }
                    else if (tablesForTransaction.Count() == 1)
                    {
                        //optimization for one table
                        //System.Diagnostics.Stopwatch sw = new System.Diagnostics.Stopwatch();
                        //sw.Start();
                        lock (_SYNC_COMMIT)
                        {                            
                            tablesForTransaction[0].Commit();                            
                        }
                        //sw.Stop();
                        //Console.WriteLine("Commit: " + sw.ElapsedMilliseconds);
                    }
                    else
                    {
                        if (ignoreTransaction)
                        {
                            lock (_SYNC_COMMIT)
                            {
                                foreach(var tft in tablesForTransaction)
                                {
                                    tft.Commit();   
                                }
                            }
                        }
                        else
                        {
                            //WORKS SLOWER THEN ONE TABLE COMMIT, because of internally integrated mechanism of transaction scope journalling.

                            STSdb.Data.Transaction internalTransaction = new STSdb.Data.Transaction(this._storageEngine, tablesForTransaction);

                            //Commit and Rollback must be serial for all threads [because of journalling file], though commits are fast and take the same time, independent from the quantity of committed data.
                            lock (_SYNC_COMMIT)
                            {
                                internalTransaction.Commit();
                            }
                        }
                    }
                }
                catch (Exception ex)
                {
                    this.UnregisterTransaction(transactionThreadId);
                    throw new Exception("STSdb.MTL: NATIVE COMMIT FAILED " + ex.ToString()); 
                }
                finally
                {
                    _sync_dbIsOperatable.ExitReadLock();
                }   
                

                //internal transaction commit
                try
                {
                    transactionUnit.Commit();
                }
                catch (Exception ex)
                {                  
                    this.StopEngineInFailureCase("In Memory Synchronization failed in Commit. \r\n" + ex.ToString());
                    return;
                }
                
            }
            else
            {
                throw new Exception("STSdb.MTL: TRANSACTION DOESN'T EXIST ANY MORE");
            }
        }

        /// <summary>
        /// Rollsback transaction
        /// </summary>
        /// <param name="transactionThreadId"></param>
        public void RollBack(int transactionThreadId)
        {

            TransactionUnit transactionUnit = this.GetTransactionUnit(transactionThreadId);

            if (transactionUnit != null)
            {

                ////////////////////////////////////////  NATIVE STSdb ROLLBACK
                _sync_dbIsOperatable.EnterReadLock();
                try
                {
                    if (!this._dbIsOperatable)
                        return;

                    STSdb.Data.Transaction internalTransaction = new STSdb.Data.Transaction(this._storageEngine, transactionUnit.GetTransactionWriteTables().Select(r => r.STSdbITransaction).ToArray());

                    //Commit and Rollback must be serial for all threads, though commits are fast and take the same time, independent from the quantity of committed data.
                    lock (_SYNC_COMMIT)
                    {
                        internalTransaction.Rollback();
                    }
                    
                }
                catch (Exception ex)
                {
                    this.UnregisterTransaction(transactionThreadId);
                    throw new Exception("STSdb.MTL: NATIVE ROLLBACK FAILED " + ex.ToString());
                }
                finally
                {
                    _sync_dbIsOperatable.ExitReadLock();
                }           

                //internal transaction rollback
                try
                {
                    transactionUnit.RollBack();
                }
                catch (Exception ex)
                {                    
                    this.StopEngineInFailureCase("In Memory Synchronization failed in Rollback. \r\n" + ex.ToString());
                    return;
                }
                
            }
            else
            {
                throw new Exception("STSdb.MTL: TRANSACTION DOESN'T EXIST ANY MORE");
            }
        }


        #region "STOP THE ENGINE"

        /// <summary>
        /// Stops the engine in case of failure
        /// </summary>
        /// <param name="exception"></param>
        public void StopEngineInFailureCase(Exception exception)
        {          
            bool dboperatable = DbIsOperatable;

            this.StopEngine();
            
            Exception e = null;

            if (dboperatable) 
                e = new Exception("STSdb.MTL: DB is not operatable anymore, reinitialze the instance. \r\n InnerException: " + exception.ToString());
            else
                e = exception;

            throw e;
        }

        /// <summary>
        /// Stops the engine in case of failure
        /// </summary>
        /// <param name="exception"></param>
        public void StopEngineInFailureCase(string exception)
        {
           this.StopEngineInFailureCase(new Exception(exception));           
        }

        /// <summary>
        /// Normal Engine Stop, usually in case of Main Thread or DLL disposing
        /// </summary>
        public void StopEngine()
        {
            _sync_dbIsOperatable.EnterWriteLock();
            try
            {
                this._dbIsOperatable = false;
            }
            finally
            {
                _sync_dbIsOperatable.ExitWriteLock();
            }

            this.UnregisterAllTransactions();
            this._storageEngine.Dispose();

        }

        #endregion
    }
}
